package com.bfxy.disruptor.heigh.multi;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

import com.bfxy.disruptor.common.EventFactory;
import com.bfxy.disruptor.common.ExceptionHandler;
import com.bfxy.disruptor.common.ProducerType;
import com.bfxy.disruptor.common.RingBuffer;
import com.bfxy.disruptor.common.SequenceBarrier;
import com.bfxy.disruptor.common.WorkerPool;
import com.bfxy.disruptor.common.YieldingWaitStrategy;

//import com.lmax.disruptor.EventFactory;
//import com.lmax.disruptor.ExceptionHandler;
//import com.lmax.disruptor.RingBuffer;
//import com.lmax.disruptor.SequenceBarrier;
//import com.lmax.disruptor.WorkerPool;
//import com.lmax.disruptor.YieldingWaitStrategy;
//import com.lmax.disruptor.dsl.ProducerType;

public class Main {

	public static void main(String[] args) throws InterruptedException {
		//不直接创建Disruptor
		
		//1 创建RingBuffer
		RingBuffer<Order> ringBuffer =
				RingBuffer.create(ProducerType.MULTI,  /*多生产者*/
						new EventFactory<Order>() {    /*消息工厂*/
							public Order newInstance() {
								return new Order();
							}
						},
						1024*1024, /*缓存大小*/ new YieldingWaitStrategy());
		
		//2 通过ringBuffer 创建一序号屏障，ProcessingSequenceBarrier
		SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
		
		//3 创建多个消费者数组:每一个消费者都有自己独立的SequenceBarrier和Sequence，
		//Sequence作用是每一个消费者消费到哪了，每一个消费者的进度，这个进度要让ringBuffer知道，
		Consumer[] consumers = new Consumer[10];
		for(int i = 0; i < consumers.length; i++) {
			consumers[i] = new Consumer("C" + i);
		}
		
		//4 构建多消费者工作池
		WorkerPool<Order> workerPool = new WorkerPool<Order>(
			ringBuffer, sequenceBarrier, new EventExceptionHandler(), consumers);
		
		//5 设置多个消费者的sequence序号的数组， 用于单独统计消费进度, 并且设置到ringbuffer中
		ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
		
		//6 启动workerPool。100个生产者10个消费者，只用了5个线程。
		workerPool.start(Executors.newFixedThreadPool(5));
		
		final CountDownLatch latch = new CountDownLatch(1);
		
		for(int i = 0; i < 100; i++) {//100个生产者，每个生产者1个线程
			final Producer producer = new Producer(ringBuffer);
			new Thread(new Runnable() {
				public void run() {
					try {
						latch.await();//每创建一个线程就阻塞着，等待信号。
					} catch (Exception e) {
						e.printStackTrace();
					}
					for(int j = 0; j<100; j++) {//每个生产者投递100个数据，公共10000条数据
						producer.sendData(UUID.randomUUID().toString());
					}
				}
			}).start();
		}
		
		Thread.sleep(2000);//线程创建完毕，都阻塞着
		System.err.println("----------线程创建完毕，都阻塞着，开始生产数据----------");
		latch.countDown();
		Thread.sleep(10000);
		System.err.println("任务总数:" + consumers[2].getCount());
	}
	
	static class EventExceptionHandler implements ExceptionHandler<Order> {
		public void handleEventException(Throwable ex, long sequence, Order event) {
		}

		public void handleOnStartException(Throwable ex) {
		}

		public void handleOnShutdownException(Throwable ex) {
		}
	}
}
